package com.muyu.kafka.consumer;

import com.muyu.common.core.utils.StringUtils;
import com.muyu.kafka.domain.Data1;
import com.muyu.kafka.domain.Data2;
import com.muyu.kafka.domain.Data3;
import com.muyu.kafka.service.KafkaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @ClassName consumer
 * @Description 描述
 * @Author LiuZeZhang
 * @Date 2024/04/11 14:10
 */
@Component
@Slf4j
public class consumer {
    @Autowired
    private KafkaService kafkaService;

    @KafkaListener(topics = "data1",containerFactory = "kafkaListenerContainerFactory",errorHandler = "myKafkaListenerErrorHandler")
    public void  consumer1(ConsumerRecord consumerRecord, Acknowledgment acknowledgment){
            Data1 value = (Data1) consumerRecord.value();
            if(StringUtils.isNull(value.getId())){
                log.info("data1id不能为空");
                return;
            }
            //查询data1中是否存在id
            Integer data1 = kafkaService.findData1(value.getId());
            if (data1==0){
                //添加
                kafkaService.insertData1(value);
            }else{
                //修改
                kafkaService.updateData1(value);
            }
            acknowledgment.acknowledge();
    }
    @KafkaListener(topics = "data2",containerFactory = "kafkaListenerContainerFactory",errorHandler = "myKafkaListenerErrorHandler")
    public void  consumer2(ConsumerRecord consumerRecord, Acknowledgment acknowledgment){
            Data2 value2 = (Data2) consumerRecord.value();
            if(StringUtils.isNull(value2.getId())){
                log.info("data2id不能为空");
                return;
            }
            //查询data2中是否存在id
            Integer data2=kafkaService.findData2(value2.getId());
            if (data2==0){
                //添加
                kafkaService.insertData2(value2);
            }else{
                //修改
                kafkaService.updateData2(value2);
            }
            acknowledgment.acknowledge();


    }

    @KafkaListener(topics = "data3",containerFactory = "kafkaListenerContainerFactory",errorHandler = "myKafkaListenerErrorHandler")
    public void  consumer3(ConsumerRecord consumerRecord, Acknowledgment acknowledgment){
            Data3 value = (Data3) consumerRecord.value();
            if(StringUtils.isNull(value.getId())){
                log.info("data3id不能为空");
                return;
            }
            //查询data3中是否存在id
            Integer data3=kafkaService.findData3(value.getId());
            if (data3==0){
                //添加
                kafkaService.insertData3(value);
            }else{
                //修改
                kafkaService.updateData3(value);
            }
            acknowledgment.acknowledge();
        }
}
